package com.zhanghe.study.springboot.rabbit;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author zh
 * @date 2021/9/25 19:37
 */
public class ProviderClient {

    public void send() {
        // 连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("");
        factory.setPassword("");

        Connection connection = null;
        Channel channel = null;
        try{
            // 创建连接
            connection = factory.newConnection();
            // 创建通道
            channel = connection.createChannel();
            // 声明队列
            // String queue,  队列名称
            // boolean durable, 是否持久化，持久化可以保证服务挂掉，而消息不会丢失
            // boolean exclusive, 是否为通道独占的
            // boolean autoDelete, 自动删除，如果为true，不使用时则自动删除
            // Map<String, Object> arguments
            channel.queueDeclare("queueName",false,false,false,null);

            // String exchange,  交换机名称
            //        BuiltinExchangeType type, 交换机类型
            //        boolean durable, 是否持久化
            //        boolean autoDelete, 是否自动删除
            //        boolean internal, 是否内部使用，true表示客户端不可以使用该交换机
            //        Map<String, Object> arguments
            channel.exchangeDeclare("",BuiltinExchangeType.DIRECT,false,false,false,null);



            // 使用direct
//            channel.exchangeDeclare("exchangeName",BuiltinExchangeType.DIRECT);

            // 发布消息
            // String exchange, exchange交换机
            // String routingKey, 路由键
            // BasicProperties props,  指定消息的基本属性
            // byte[] body  消息体
            channel.basicPublish("test","qu",null,"我发送消息了".getBytes());


            // 生产者如果需要知道消息是否已经发送到rabbitMQ，则需要rabbitMQ进行反馈
            // 有两种方式
            // ①confirm机制
            channel.confirmSelect();
            channel.waitForConfirms();

            // ②事务机制
            channel.txSelect();
            // 事务回滚
            channel.txRollback();
        } catch (Exception e){
            
        } finally {
            try{
                channel.close();

            } catch (IOException | TimeoutException e){

            }

            try{
                connection.close();

            } catch (IOException e){

            }

        }
        
    }

    public void receive() {
        // 连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("");
        factory.setPassword("");

        Connection connection = null;
        Channel channel = null;
        try{
            // 创建连接
            connection = factory.newConnection();
            // 创建通道
            channel = connection.createChannel();

            // 绑定队列和交换机
            // String queue,  队列名
            // String exchange, 交换机名
            // String routingKey, 路由键
            // Map<String, Object> arguments
            channel.queueBind("queueName","exchangeName","routingKey",null);

            // 如果进行配置，则该消费者只能处理一条消息
            // 限定条数
            channel.basicQos(1);

            // 消息确认，消费者发送一个确认信息来告诉rabbitMQ，已经接收到消息并且处理完毕，此时rabbitMQ就可以删除消息了
            // 如果一个消费者在发送确认消息之前挂掉，rabbitMQ会认为该消息没有被接收处理，会重新将消息放入到队列中
            // autoAck为true，为自动确认，则表示接收到消息就会自动确认已经接收并处理了，消息删除
            // autoAck为false，为手动确认，则需要通过显式的调用channel.basicAck(envelope.getDeliveryTag(),false);来通知rabbitMQ
            // 消息已接收并处理，可以进行消息删除了

            // 消费者接收处理
            // String queue,  队列名
            // boolean autoAck, 自动确认
            // Consumer callback 消费者回调
            channel.basicConsume("queueName", true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, // 消费者标签
                                           Envelope envelope, // 信封
                                           AMQP.BasicProperties properties, // 基本信息
                                           byte[] body)
                        throws IOException
                {
                    String message = new String(body, StandardCharsets.UTF_8);

                    // 手动拒绝
                    //  requeue 表示是否重新入队列，false表示直接丢弃，true表示重新加入队列
                    this.getChannel().basicReject(envelope.getDeliveryTag(),false);

                }
            });



        } catch (Exception e){

        } finally {
            try{
                channel.close();

            } catch (IOException | TimeoutException e){

            }

            try{
                connection.close();

            } catch (IOException e){

            }

        }

    }
}
